package cn.itcast.up.model.statistisc

import cn.itcast.up.model.base.BaseModel
import cn.itcast.up.model.statistisc.CycleModel.{computeTagIds, getFiveDF, getFourRule, getMysqlSource, loadHBaseData, loadOldData, mergeTotalTag, sinkData, spark}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.expressions.Window

/**
  * 常用支付方式标签计算
  */
object PaymentModel extends BaseModel{
  import spark.implicits._
  import org.apache.spark.sql.functions._

  /**
    * 设置当前应用的名字
    *
    * @return
    */
  override def setAppName(): String = "PaymentModel"

  /**
    * 设置4级标签ID
    *
    * @return
    */
  override def setFourTagID(): Int = 508

  /**
    * 标签计算
    *
    * @param hbaseDF HBase数据源
    * @param fiveDF  规则数据
    * @return 最终的计算结果:userid, tagIds
    */
  override def computeTagIds(hbaseDF: DataFrame, fiveDF:  DataFrame): DataFrame = {
    //将5级数据转换为Map
    val ruleMap: Map[String, String] = fiveDF.map(row => {
      val id: String = row.getAs[Long]("id").toString
      val rule: String = row.getAs[String]("rule").toString
      //(规则,标签ID)
      (rule, id)
    }).collect().toMap

//    fiveDF.show()
//    +---+---------------+
    //| id|           rule|
    //+---+---------------+
    //|509|         alipay|
    //|510|         kjtpay|
    //|511|            cod|
    //|512|insidestatement|
    //|513|            ccb|
    //|514|       chinapay|
    //|515|        prepaid|
    //|516|          other|
    //+---+---------------+
//    hbaseDF.show()
//+---------+-----------+
    //| memberId|paymentCode|
    //+---------+-----------+
    //| 13823431|     alipay|
    //|  13823431|     alipay|
    //|  13823431|     alipay|
    //|  4035041|     alipay|
    //| 13823431|     kjtpay|
    //|  4034219|     alipay|
    //|138230939|     alipay|
//    hbaseDF.createOrReplaceTempView("paymethod")

   /* val sql =
      """
        |
        |select *,row_number() over(distribute by memberId order by counts desc) rn from (
        |select memberId, paymentCode, count(paymentCode) counts from paymethod group by memberId,paymentCode) t1
        |
      """.stripMargin*/

//   select memberId, paymentCode, count(paymentCode) counts from paymethod group by memberId,paymentCode
    //    +--------+-----------+------+
    //|memberId|paymentCode|counts|
    //+--------+-----------+------+
    //|13823481|     alipay|    96|
    //| 4035297|     alipay|    80|
    //|13823317|     kjtpay|    11|
    //|13822857|     alipay|   100|

//
    //    select *,row_number() over(distribute by memberId order by counts desc) rn from (
    //    select memberId, paymentCode, count(paymentCode) counts from paymethod group by memberId,paymentCode) t1
    //    +---------+-----------+------+---+
    //| memberId|paymentCode|counts| rn|
    //+---------+-----------+------+---+
    //| 13822725|     alipay|    89|  1|
    //| 13822725|        cod|    12|  2|
    //| 13822725|     kjtpay|     9|  3|
    //| 13822725|      wspay|     3|  4|
    //| 13822725|   giftCard|     2|  5|
    //| 13822725|    prepaid|     1|  6|

//    spark.sql(sql)
//      .show()
    //获取用户常用的支付方式
    val getTagId = udf((paymentCode: String) => {
  //获取用户支付方式ID,如果找不到,就返回一个other
      ruleMap.getOrElse(paymentCode, ruleMap.getOrElse("other", ""))
    })


    //使用DSL统计用户最常用的支付方式
    //分组
    hbaseDF.groupBy('memberId, 'paymentCode)
      //统计
      .agg(count('paymentCode).as("counts"))
    //在之前的数据之上添加一列:rn
      .withColumn("rn",
      row_number().over(Window.partitionBy('memberId).orderBy('counts.desc)))
//      .show()
//+---------+-----------+------+---+
    //| memberId|paymentCode|counts| rn|
    //+---------+-----------+------+---+
    //| 13822725|     alipay|    89|  1|
    //| 13822725|        cod|    12|  2|
    //| 13822725|     kjtpay|     9|  3|
    //| 13822725|      wspay|     3|  4|
    //| 13822725|   giftCard|     2|  5|
    //| 13822725|    prepaid|     1|  6|
      //获取最常用的支付方式
      .where("rn = 1")
      //对5级规则数据进行关联,获取对应的ID
//      .show()
//+---------+-----------+------+---+---+------+
    //| memberId|paymentCode|counts| rn| id|  rule|
    //+---------+-----------+------+---+---+------+
    //| 13822691|        cod|     2|  1|511|   cod|
    //| 13822725|     alipay|    89|  1|509|alipay|
    //| 13823083|     alipay|    94|  1|509|alipay|
      .select('memberId.as("userid"), getTagId('paymentCode).as("tagIds"))
  }

  def main(args: Array[String]): Unit = {
    //加载MySQL数据源
    val mysqlSource: DataFrame = getMysqlSource
    //获取4级规则的map
    val map: Map[String, String] = getFourRule(mysqlSource)
    //获取5级规则的DataFrame
    val fiveDF: DataFrame = getFiveDF(mysqlSource)
    //加载HBase数据源
    val hbaseSource: DataFrame = loadHBaseData(map)
    //开始进行标签计算
    val newDF: DataFrame = computeTagIds(hbaseSource, fiveDF)
    //加载历史数据
    val oldDF: DataFrame = loadOldData(map)
    //进行标签合并
    val result: DataFrame = mergeTotalTag(newDF, oldDF)
    //将合并之后的数据落地
    sinkData(result, map)
  }
}
